package com.gator.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.UUID;


public class PABLO_KafkaProducer extends Thread {

    //生产者对象
    KafkaProducer<Integer, String> producer;
    //主题
    String topic;

    public PABLO_KafkaProducer(String topic) {
        Properties properties = new Properties();
        //bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "localhost:9092,localhost:9093,localhost:9094");
        //client.id
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "pablo-producer");
        //partitioner.class 关联自定义分区器，默认是default
       /* properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,
                "com.pablo_kafka.kafka.PABLO_MyPartition");*/
        //key 和 value的序列化
        //org.apache.kafka.common.serialization.IntegerSerializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                IntegerSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        //----------------优化参数------------------
        //分区器缓冲区大小默认32M
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 1024L * 1024 * 32 * 2);
        //默认16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384 * 2);
        //默认sendThread等待0ms
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 50);
        //配置压缩,默认none
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        //-----------------------------------------
        //应答策略  默认-1all ,开始事务这里必须是-1/all
        properties.put(ProducerConfig.ACKS_CONFIG, "-1");
        //重试次数,即sender线程重试次数，即集群返回失败后重试，默认int最大值，死磕~~
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        //事务必须手动指定事务id
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, UUID.randomUUID().toString());

        //通过配置文件初始化生产者
        producer = new KafkaProducer<Integer, String>(properties);
        //初始化topic
        this.topic = topic;


    }

    @Override
    public void run() {

        //初始化事务
        producer.initTransactions();
        //开启事务
        producer.beginTransaction();
        try {
            for (int i = 0; i < 20; i++) {
                //普通异步发送
                //producer.send(new ProducerRecord<>(topic,msg));

                //异步回调，返回发送的结果
                producer.send(new ProducerRecord<>(topic, "hello" + i), (metadata, exception) -> {
                    //打印回调的结果
                    if (exception == null) {
                        System.out.println("偏移量：" + metadata.offset() + "   分区是："
                                + metadata.partition() + "   主题是："
                                + metadata.topic());
                    }
                });
                Thread.sleep(2000);
            }
            //提交事务
            producer.commitTransaction();

        } catch (Exception e) {
            e.printStackTrace();
            //关闭事务
            producer.abortTransaction();
        } finally {
            producer.close();
        }
    }

    public static void main(String[] args) {
        new PABLO_KafkaProducer("test_partitions").start();
    }
}
